package com.unisound.tracelog.server.mq.consumer;

import cn.hutool.core.date.LocalDateTimeUtil;
import com.unisound.tracelog.common.model.span.Span;
import com.unisound.tracelog.common.utils.JsonUtils;
import com.unisound.tracelog.server.model.entity.TraceLog;
import com.unisound.tracelog.server.service.TraceLogService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

@Slf4j
@Component
public class TraceLogKafkaConsumer {

    @Autowired
    private TraceLogService traceLogService;

    @KafkaListener(topics = {"${tracelog.kafka.consumer.span.topic}"}, groupId = "${tracelog.kafka.consumer.span.group-id}")
    public void consume(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
        List<TraceLog> traceLogs = new ArrayList<>();
        for (ConsumerRecord<String, String> record : records) {
            String value = record.value();
            Span span = JsonUtils.fromJson(value, Span.class);
            if (span == null) continue;

            traceLogs.add(TraceLog.builder()
                    .traceId(span.getTraceId())
                    .spanId(span.getSpanId())
                    .parentSpanId(span.getParentSpanId())
                    .startTime(span.isInput() ? LocalDateTimeUtil.of(span.getExecTime()) : null)
                    .endTime(span.isInput() ? null : LocalDateTimeUtil.of(span.getExecTime()))
                    .input(span.isInput() ? span.getContent() : null)
                    .output(span.isInput() ? null : span.getContent())
                    .resultFlag(span.getResultFlag())
                    .name(span.getName())
                    .serviceName(span.getServiceName())
                    .type(span.getType())
                    .elapsedTime(span.getElapsedTimeMillis())
                    .build());
        }

        if (CollectionUtils.isNotEmpty(traceLogs)) {
            traceLogService.saveOrUpdateBatch(traceLogs);
        }

        try {
            ack.acknowledge();
        } catch (Exception e) {
            log.error("tracelog消费异常", e);
        }
    }

}
